package com.atguigu.flinkCEP;

import com.atguigu.been.LoginEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author wky
 * @create 2021-07-20-15:54
 */
//2秒内连续登录失败2次及以上的用户
public class Flink11_CEP_Project_Login {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> streamSource =senv.readTextFile("src/input/LoginLog.csv");

        WatermarkStrategy<LoginEvent> watermarkStrategy = WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(50))
                .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                    @Override
                    public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                        return element.getEventTime();
                    }
                });

        KeyedStream<LoginEvent, Long> loginEventLongKeyedStream = streamSource.map(new MapFunction<String, LoginEvent>() {
            @Override
            public LoginEvent map(String value) throws Exception {
                String[] split = value.split(",");
                return new LoginEvent(Long.valueOf(split[0]),
                        split[1],
                        split[2],
                        Long.parseLong(split[3]) * 1000L);
            }
        })
                .assignTimestampsAndWatermarks(watermarkStrategy)
                .keyBy(LoginEvent::getLoginId);

        Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("login")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return "fail".equals(value.getEventType());
                    }
                })
                .times(2)
                .consecutive()
                .within(Time.seconds(2));
        PatternStream<LoginEvent> patternStream = CEP.pattern(loginEventLongKeyedStream, pattern);
        patternStream.select(new PatternSelectFunction<LoginEvent, String>() {
            @Override
            public String select(Map<String, List<LoginEvent>> map) throws Exception {
                return map.toString() ;
            }
        }).print();
        senv.execute();

    }
}
